package com.example.status;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created with IntelliJ IDEA.
 * ClassName: OperatorListStatusDemo
 * Package: com.example.status
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-21
 * Time: 22:04
 */

//状态算子
public class OperatorListStatusDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //在map算子中 计算数据的个数
        env.socketTextStream("hadoop103",7777)
                //map本身是无状态算子 现在在里面定义状态
                .map(new MyCountMapFunction())
                .print();





        env.execute();

    }

    //1.实现CheckpointedFunction 接口
    public static class MyCountMapFunction implements MapFunction<String,Long>, CheckpointedFunction {

        private Long count;

        @Override
        public Long map(String value) throws Exception {
            return count++;
        }

        /**
         * 本地变量的持久化 定义 将 本地变量 拷贝到 算子状态中
         * @param context
         * @throws Exception
         */
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {

        }

        /**
         * 初始化本地变量 从状态中 把数据添加到 本地变量 每个子任务调用一次
         * @param context
         * @throws Exception
         */
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {

        }
    }

}
